package com.juma.cartridge.goodevent;

import java.io.IOException;
import java.io.Serializable;
import java.rmi.RemoteException;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;

import com.juma.cartridge.common.HessianIOUtils;
import com.juma.cartridge.common.RetryExecutor;
import com.juma.cartridge.common.RetryUtils;
import com.juma.cartridge.retryque.RetryHandler;
import com.juma.cartridge.retryque.RetryQueueAdmin;
import com.juma.cartridge.retryque.RetryQueueManager;
import com.juma.cartridge.retryque.RetryResult;
import com.juma.cartridge.retryque.RetryTask;
import com.juma.cartridge.retryque.TaskStatus;
import com.third.service.boot.starter.eventbus.domain.Event;

/**
 * 对eventbus发送客户端进行增强,结合retryque实现可靠发送
 * @ClassName EventBusEnhancer
 * @Description
 * @Author zhuangguangyou
 * @Date 2019年10月16日 上午11:28:51
 * @Version 1.0
 *
 */
@Slf4j
public final class EventBusEnhancer implements RetryHandler,RabbitTemplate.ConfirmCallback {

    @Resource
    private RetryQueueManager retryQueueManager;
    
    @Resource
    private RetryQueueAdmin retryQueueAdmin;
    
    @Resource
    private RabbitTemplate rabbitTemplate;
    
    private final String bizCode = "eventSender";

    private EventPushListener pushListener;
    
    private int maxRetries = 300; //最大重试次数
    
    @PostConstruct
    public void init() {
        CachingConnectionFactory ccf = (CachingConnectionFactory)rabbitTemplate.getConnectionFactory();
        ccf.setPublisherConfirms(true);
        rabbitTemplate.setConfirmCallback(this);
        retryQueueManager.registerRetryHandler(bizCode, this);
    }
    
    /**
     * 发送事件(异步)
     * @param eventId 事件ID
     * @param data 事件数据
     */
    public void push(Event<Serializable> event) {
        
        if(event.getEventId()==null||event.getValue()==null) {
            throw new IllegalArgumentException("eventId或value为空");
        }
        
        RetryTask task = new RetryTask();
        task.setBizCode(bizCode);
        task.setBizFlag(event.getEventId());
        try {
            byte[] bytes = HessianIOUtils.serialize(event.getValue());
            task.setBizBin(bytes);    
        }catch(IOException ioe) {
            throw new IllegalArgumentException("数据序列化失败", ioe);
        }
        retryQueueManager.enqTask(task);
    }

    
    
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        //收到消息回调
        final Long taskId = Long.parseLong(correlationData.getId());
        final Set<Long> ids = new TreeSet<Long>();
        ids.add(taskId);
        
        int recallTimes = 3;
        int interval = 1000;
        if(ack) {
            
            //回调消费成功监听接口
            if(pushListener!=null) {
                
                RetryUtils.retry(new RetryExecutor<Boolean>(){
                    @Override
                    public Boolean execute() {
                        try {
                            Map<Long,RetryTask> tasks = retryQueueAdmin.getMultiTasks(ids);
                            if(tasks.isEmpty()) {
                                return null;
                            }
                            RetryTask task = tasks.get(taskId);
                            pushListener.onSucceed(buildEvent(task));
                        }catch(Exception e2) {
                            log.error("回调pushListener#onSucceed接口异常", e2);
                        }
                        return Boolean.TRUE;
                    }
                    
                },recallTimes,interval);
            }
            
            //消息被Broker接收成功,删除本地消息(不再重试)
            RetryUtils.retry(new RetryExecutor<Boolean>(){
                @Override
                public Boolean execute() {
                    return retryQueueAdmin.clearTask(ids);
                }
            },recallTimes,interval);
            
        } else {
            //消息被Broker处理失败,则恢复重试
            RetryUtils.retry(new RetryExecutor<Boolean>(){
                @Override
                public Boolean execute() {
                    Map<Long,RetryTask> tasks = retryQueueAdmin.getMultiTasks(ids);
                    RetryTask task = tasks.get(taskId);
                    //如果任务还在处理中,则等待其结束后再操作...
                    if(task.getStatus()==TaskStatus.DELAY) {
                        return retryQueueAdmin.recoverTask(ids);    
                    }
                    return null;
                }
            },30,100);
        }
        
        //PS:
        //上面的代码中用到了executeReCall方法
        //这样设计的目的是:假设消息存储到本地任务表可能会有一定的延时,
        //当确MQ回调confirm确认消息时,消息此时有可能还未入库,所以confirm线程做一定时长的等待.
        //(这是一种保证本地消息被确认的手段)
    }

    /**
     * 将任务代表的事件消息进行发送
     * @param task
     * @throws IllegalStateException
     * @throws RemoteException
     */
    protected void sendMessage(RetryTask task,Event<Serializable> message) throws RemoteException {
        
        Event<Serializable> event = message;
        if(event==null) {
            event = buildEvent(task);
        }
        try {
            CorrelationData cd = new CorrelationData(String.valueOf(task.getId()));
            rabbitTemplate.convertAndSend(event.getClass().getName(), "", event, cd);    
        }catch(Exception e) {
            throw new RemoteException("发送消息到MQ服务器异常:"+e.getMessage(),e);
        }
    }

    
    @Override
    public RetryResult retry(RetryTask task) {
        if(task.getRetries()>this.maxRetries) {
            task.setDetail("重试次数超限");
            return RetryResult.BROKEN;
        }
        try {
            //发送消息
            sendMessage(task,null);
            //发送时没报错,则中止下一次重试(由confirm回调方法来决定是恢复重试还是结速任务)
            return RetryResult.DELAY;
        }catch(RemoteException e1) {
            //如果产生消息异常,则保持重试
            task.setDetail(e1.getMessage());
            return RetryResult.RETRY;
        }catch(RuntimeException e) {
            task.setDetail(e.getMessage());
            //如果产生其它异常,则重试5次后中止
            if(task.getRetries()<5) {
               return RetryResult.RETRY;
            }
            
            if(pushListener!=null) {
                try {
                    pushListener.onFailed(buildEvent(task), e);    
                }catch(Exception e2){
                    log.error("回调pushListener#onFailed接口异常", e2);
                }
            }
            return RetryResult.BROKEN;
        }
    }
    
    public void setPushListener(EventPushListener pushListener) {
        this.pushListener = pushListener;
    }

    private Event<Serializable> buildEvent(RetryTask task) {
        Event<Serializable> event = new Event<Serializable>();
        event.setEventId(task.getBizFlag());
        try {
            event.setValue((Serializable)HessianIOUtils.deserialize(task.getBizBin()));        
        }catch(IOException ioe) {
            throw new IllegalStateException("从本地消息表中反序列化数据失败:"+ioe.getMessage(), ioe);
        }
        return event;
    }
    

    public void setRetryQueueManager(RetryQueueManager retryQueueManager) {
        this.retryQueueManager = retryQueueManager;
    }

    public void setRetryQueueAdmin(RetryQueueAdmin retryQueueAdmin) {
        this.retryQueueAdmin = retryQueueAdmin;
    }

    public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void setMaxRetries(int maxRetries) {
        this.maxRetries = maxRetries;
    }
    
}
